package com.heima.article.kafkaStream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.common.mess.ArticleVisitStreamMess;
import com.heima.common.mess.UpdateArticleMess;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

/**
 * kafkastream的配置类，用来进行流式计算
 */
@Configuration
public class KSteamConfig {

    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
        //1.接收生产者发送的消息
        KStream<String, String> kStream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);
        //2.处理消息
        kStream.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
                    /**
                     *
                     * @param key  表示发送消息的key ,但是此时为空
                     * @param value 此时是一个json字符串 “{'articleId':123123,'type':'views','add':1}”
                     * @return
                     *      返回值    key=文章id   value=type:add  相当于views:1
                     */
                    @Override
                    public KeyValue<String, String> apply(String key, String value) {
                        //对value进行转化类型
                        UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
                        return new KeyValue<>(mess.getArticleId().toString(), mess.getType() + ":" + mess.getAdd());
                    }
                })
                /**
                 *  分组
                 *  此时key= 文章id
                 *  此时的value= views:1
                 */
                .groupBy((key, value) -> key)
                //每10s拉取一次
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                //聚合操作
                .aggregate(new Initializer<String>() {
                               /**
                                * 初始化方法
                                * 我们最终达到的目标是 "view:1,collect:1,comment:1,like:1"
                                */
                               @Override
                               public String apply() {
                                   return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                               }
                           },
                        /**
                         * 属于真正的聚合操作
                         * key=文章id
                         * value=views:1
                         * init表示初始化中的返回值 "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0"
                         */
                        new Aggregator<String, String, String>() {
                            @Override
                            public String apply(String key, String value, String init) {
                                //1.判断value是否为空
                                if (StringUtils.isBlank(value)) {
                                    return init;
                                }
                                //定义变量
                                int collection = 0, comment = 0, like = 0, view = 0;
                                //对value进行分割
                                String[] split = value.split(":");//split[0]=views,split[1]=1
                                switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])) {
                                    case LIKES:
                                        like += Integer.parseInt(split[1]);
                                        break;
                                    case COLLECTION:
                                        collection += Integer.parseInt(split[1]);
                                        break;
                                    case COMMENT:
                                        comment += Integer.parseInt(split[1]);
                                        break;
                                    case VIEWS:
                                        view += Integer.parseInt(split[1]);
                                        break;
                                }
                                //拼接字符串
                                String format = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", collection, comment, like, view);
                                return format;
                            }
                        },
                        Materialized.as("aggregatename"))
                //转化类型为kstream
                .toStream()
                //准备发送消息内容
                .map(new KeyValueMapper<Windowed<String>, String, KeyValue<String, String>>() {
                    /**
                     *
                     * @param key 文章id  表示发消息的key
                     * @param value "COLLECTION:3,COMMENT:2,LIKES:1,VIEWS:1" 发送消息的内容
                     * @return 返回结果的消息内容中必须包含文章id
                     */
                    @Override
                    public KeyValue<String, String> apply(Windowed<String> key, String value) {
                        return new KeyValue<String, String>(key.key(), formatValue(key.key(), value));
                    }
                })
                //发送消息
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);
        return kStream;
    }
    /**
     * 格式化消息内容
     *
     * @param key   文章id  表示发消息的key
     * @param value "COLLECTION:3,COMMENT:2,LIKES:1,VIEWS:1" 发送消息的内容
     * @return
     */
    private String formatValue(String key, String value) {
        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
        mess.setArticleId(Long.parseLong(key));
        String[] aggArray = value.split(",");//agg[0]="COLLECTION:3",agg[1]="COMMENT:2"
        for (String agg : aggArray) {
            String[] split = agg.split(":");//split[0]=views,split[1]=1
            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])) {
                case LIKES:
                    mess.setLike(Integer.parseInt(split[1]));
                    break;
                case COLLECTION:
                    mess.setCollect(Integer.parseInt(split[1]));
                    break;
                case COMMENT:
                    mess.setComment(Integer.parseInt(split[1]));
                    break;
                case VIEWS:
                    mess.setView(Integer.parseInt(split[1]));
                    break;
            }
        }
        return JSON.toJSONString(mess);
    }
}
